package org.apache.rocketmq.proxy;

import com.google.common.collect.Lists;
import io.grpc.BindableService;
import io.grpc.protobuf.services.ChannelzService;
import io.grpc.protobuf.services.ProtoReflectionService;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.BrokerStartup;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.common.utils.AbstractStartAndShutdown;
import org.apache.rocketmq.proxy.config.Configuration;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.grpc.GrpcServer;
import org.apache.rocketmq.proxy.grpc.GrpcServerBuilder;
import org.apache.rocketmq.proxy.grpc.v2.GrpcMessagingApplication;
import org.apache.rocketmq.proxy.processor.DefaultMessagingProcessor;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.proxy.remoting.RemotingProtocolServer;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.srvutil.ServerUtil;

import java.util.Date;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static org.apache.rocketmq.broker.BrokerStartup.createBrokerController;

public class ProxyStartup {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);

    private static ProxyStartAndShutdown PROXY_START_AND_SHUTDOWN = new ProxyStartAndShutdown();
    private static class ProxyStartAndShutdown extends AbstractStartAndShutdown {

        @Override
        public void appendStartAndShutdown(StartAndShutdown startAndShutdown) {
            super.appendStartAndShutdown(startAndShutdown);
        }
    }
    public static void main(String[] args) {
        try {
            //从命令行参数中解析用户配置的参数信息，在我为大家提供的测试类中
            //我只配置了proxy程序的部署启动方式，我设置成local模式了，也就是本地模式
            //具体的命令为-pm local
            CommandLineArgument commandLineArgument = parseCommandLineArgument(args);
            //根据用户配置的命令行菜属初始化proxy模块的配置信息
            initConfiguration(commandLineArgument);
            //初始化线程池监控组件
            initThreadPoolMonitor();
            //创建消息处理器，proxy模块接收到的所有客户端请求最后都是由消息处理器处理的
            MessagingProcessor messagingProcessor = createMessagingProcessor();
            //创建Remoting协议服务器，其实就是创建了Netty构建的服务端，proxy模块提供两个服务端
            //一个是grpc服务端，一个是netty服务端，在第三版本代码中，我只为大家把Netty服务端构建好了
            RemotingProtocolServer remotingServer = new RemotingProtocolServer(messagingProcessor);
            //remotingServer对象还实现了StartAndShutdown接口，所以这个对象同时又是一个启动和关闭管理器
            //因此这个remotingServer对象可以添加到PROXY_START_AND_SHUTDOWN对象中，这样一来当PROXY_START_AND_SHUTDOWN对象的start方法被调用时
            //Netty构建的服务器也就启动了，而当PROXY_START_AND_SHUTDOWN对象的shutdown方法被调用时，Netty构建的服务器也就关闭了
            PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(remotingServer);
            //创建Grpc服务器要使用的线程池
            ThreadPoolExecutor executor = createServerExecutor();
            //创建Grpc服务器，ConfigurationManager.getProxyConfig().getGrpcServerPort()得到Grpc服务器的端口号，这里得到的就是8081
            GrpcServer grpcServer = GrpcServerBuilder.newBuilder(executor, ConfigurationManager.getProxyConfig().getGrpcServerPort())
                    //创建Grpc服务器的消息处理器，Grpc服务器接收到的客户端消息都会先交给这个消息处理器处理
                    //这里也可以看到，createServiceProcessor()方法会返回一个GrpcMessagingApplication消息处理器对象
                    //而这个消息处理器对象还持有了之前创建的messageProcessor消息处理器对象
                    //在第三版本代码中我们已经知道了，不管是什么消息，最后都是要交给messageProcessor消息处理器处理
                    //现在看来，我们创建的Grpc服务器的GrpcMessagingApplication消息处理器也只是对messageProcessor消息处理器做一层代理
                    //真正处理消息，还是要交给messageProcessor消息处理器处理
                    .addService(createServiceProcessor(messagingProcessor))
                    //设置Grpc通道的监控服务Channelz 公开了四种访问 channelz 数据的方法：
                    //
                    //GetTopChannels 返回通道树的根通道
                    //GetServers 返回系统中的服务器
                    //GetChannel 返回有关特定频道的信息
                    //GetSubchannel 返回有关特定子频道的信息
                    //GetSocket 返回有关套接字的信息
                    .addService(ChannelzService.newInstance(100))
                    //设置协议信息描述服务
                    .addService(ProtoReflectionService.newInstance())
                    //设置拦截器
                    .configInterceptor()
                    //设置Grpc服务器的关闭时间
                    .shutdownTime(ConfigurationManager.getProxyConfig().getGrpcShutdownTimeSeconds(), TimeUnit.SECONDS)
                    //得到Grpc服务器
                    .build();
            //把Grpc服务器添加到启动和关闭管理器中，这样一来Grpc服务器也可以通过启动和关闭管理器来启动和关闭
            PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(grpcServer);








            //启动PROXY_START_AND_SHUTDOWN对象，这个时候Netty构建的服务器也就启动了
            PROXY_START_AND_SHUTDOWN.start();
            //向JVM注册一个钩子函数，JVM关闭的时候会调用这个钩子函数，这里就是在JVM关闭的时候执行了PROXY_START_AND_SHUTDOWN对象的shutdown方法
            //这样一来Netty构建的服务器也就关闭了
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                log.info("try to shutdown server");
                try {
                    PROXY_START_AND_SHUTDOWN.preShutdown();
                    PROXY_START_AND_SHUTDOWN.shutdown();
                } catch (Exception e) {
                    log.error("err when shutdown rocketmq-proxy", e);
                }
            }));


        } catch (Exception e) {
            e.printStackTrace();
            log.error("find an unexpect err.", e);
            System.exit(1);
        }
        System.out.printf("%s%n", new Date() + " rocketmq-proxy startup successfully");
        log.info(new Date() + " rocketmq-proxy startup successfully");
    }

    private static GrpcMessagingApplication createServiceProcessor(MessagingProcessor messagingProcessor) {
        //创建Grpc消息应用处理器
        GrpcMessagingApplication application = GrpcMessagingApplication.create(messagingProcessor);
        PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(application);
        return application;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/2
     * @方法描述：创建Grpc服务器的要使用的线程池的方法
     */
    public static ThreadPoolExecutor createServerExecutor() {
        //得到Proxy模块的配置信息对象
        ProxyConfig config = ConfigurationManager.getProxyConfig();
        //得到配置的线程池中线程的数量
        int threadPoolNums = config.getGrpcThreadPoolNums();
        //得到线程池任务队列的容量
        int threadPoolQueueCapacity = config.getGrpcThreadPoolQueueCapacity();
        //创建线程池
        ThreadPoolExecutor executor = ThreadPoolMonitor.createAndMonitor(
                threadPoolNums,
                threadPoolNums,
                1, TimeUnit.MINUTES,
                "GrpcRequestExecutorThread",
                threadPoolQueueCapacity
        );//JVM关闭的时候回调线程池的shutdown方法，关闭线程池
        PROXY_START_AND_SHUTDOWN.appendShutdown(executor::shutdown);
        return executor;
    }
    private static MessagingProcessor createMessagingProcessor() {
        //得到Proxy模块的启动方式
        String proxyModeStr = ConfigurationManager.getProxyConfig().getProxyMode();
        DefaultMessagingProcessor messagingProcessor ;
        if (ProxyMode.isClusterMode( proxyModeStr)){
            messagingProcessor =  DefaultMessagingProcessor.createForClusterMode();
        }else if (ProxyMode.isLocalMode(proxyModeStr)){
            //如果是本地模式启动Proxy模块的话，那么Broker是集成在Proxy中的
            //Broker会随着Proxy的启动而启动，所以在这里把Broker控制器创建出来
            BrokerController brokerController = createBrokerController();
            //创建Broker控制器的启动和关闭包装类
            StartAndShutdown brokerControllerWrapper = new StartAndShutdown() {
                @Override
                public void start() throws Exception {
                    //Proxy启动的时候，Broker控制器也启动
                    brokerController.start();
                    //记录Broker启动成功的日志信息
                    String tip = "The broker[" + brokerController.getBrokerConfig().getBrokerName() + ", "
                            + brokerController.getBrokerAddr() + "] boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
                    if (null != brokerController.getBrokerConfig().getNamesrvAddr()) {
                        tip += " and name server is " + brokerController.getBrokerConfig().getNamesrvAddr();
                    }
                    log.info(tip);
                }

                @Override
                public void shutdown() throws Exception {
                    brokerController.shutdown();
                }
            };
            //把Broker控制器的启动和关闭包装类添加到启动和关闭管理器中
            PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(brokerControllerWrapper);

            //如果是本地模式则创建本地模式下使用的消息处理器
            messagingProcessor = DefaultMessagingProcessor.createForLocalMode(brokerController);
        }else{
             throw new IllegalArgumentException("Unsupported proxy mode: " + proxyModeStr);
        }
        PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(messagingProcessor);
         return messagingProcessor;
    }


    private static void initConfiguration(CommandLineArgument commandLineArgument) throws Exception {
        //先从命令行参数中检查用户是否配置了proxy模块的配置文件路径
        if (StringUtils.isNotBlank(commandLineArgument.getProxyConfigPath())) {
            //如果用户设置了配置路径，那就先把配置路径设置到系统属性中
            System.setProperty(Configuration.CONFIG_PATH_PROPERTY, commandLineArgument.getProxyConfigPath());
        }
        //使用配置信息管理器初始化环境变量信息
        ConfigurationManager.initEnv();
        //初始化Proxy的配置信息，这个操作执行完毕之后，ProxyConfig对象就初始化完成了，里面的各个成员变量就赋值完毕了
        ConfigurationManager.intConfig();
        //在这里使用命令行参数信息更新ProxyConfig对象的信息
        setConfigFromCommandLineArgument(commandLineArgument);
        log.info("Current configuration: " + ConfigurationManager.formatProxyConfig());
    }
    protected static BrokerController createBrokerController() {
        ProxyConfig config = ConfigurationManager.getProxyConfig();
        //在这里得到Broker控制器的启动参数列表，其实就是Broker控制器的配置文件路径
        List<String> brokerStartupArgList = Lists.newArrayList("-c", config.getBrokerConfigPath());
        //如果设置了Namesrv地址，那么就把Namesrv地址设置到Broker控制器的启动参数列表中
        if (StringUtils.isNotBlank(config.getNamesrvAddr())) {
            brokerStartupArgList.add("-n");
            brokerStartupArgList.add(config.getNamesrvAddr());
        }
        //将Broker控制器的启动参数列表转换为数组
        String[] brokerStartupArgs = brokerStartupArgList.toArray(new String[0]);
        //在这里创建Broker控制器
        return BrokerStartup.createBrokerController(brokerStartupArgs);
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/31
     * @方法描述：把用户设置的命令行参数信息设置到ProxyConfig对象中
     */
    private static void setConfigFromCommandLineArgument(CommandLineArgument commandLineArgument) {
        //设置Namesrv地址
        if (StringUtils.isNotBlank(commandLineArgument.getNamesrvAddr())) {
            ConfigurationManager.getProxyConfig().setNamesrvAddr(commandLineArgument.getNamesrvAddr());
        }
        //设置Broker配置文件路径
        if (StringUtils.isNotBlank(commandLineArgument.getBrokerConfigPath())) {
            ConfigurationManager.getProxyConfig().setBrokerConfigPath(commandLineArgument.getBrokerConfigPath());
        }
        //设置Proxy模块启动模式
        if (StringUtils.isNotBlank(commandLineArgument.getProxyMode())) {
            ConfigurationManager.getProxyConfig().setProxyMode(commandLineArgument.getProxyMode());
        }
    }


    /*
     * @description: 根据用户配置的命令行参数，初始化proxy模块的配置信息
     */
    private static CommandLineArgument parseCommandLineArgument(String[] args) {
        //解析命令行参数信息，得到了封装参数信息的CommaLine对象
        CommandLine commandLine = ServerUtil.parseCmdLine("mqproxy", args, buildCommandlineOptions(), new DefaultParser());
        if (commandLine == null) {
            throw new RuntimeException("parse command line argument failed");
        }
        //创建CommandLineArgument对象，这个对象会封装命令行参数信息，这个对象就是Proxy模块自己使用的
        CommandLineArgument commandLineArgument = new CommandLineArgument();
        //把commandLine命令行参数信息转换为Properties对象，然后再使用反射的方式把Properties对象的信息设置到CommandLineArgument对象中
        MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), commandLineArgument);
        return commandLineArgument;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/31
     * @方法描述：构建命令行有效参数的方法
     */
    private static Options buildCommandlineOptions() {
        //创建Options对象，这个对象会封装多个Option对象
        //而每一个Option对象都会封装一个命令行信息，也就是说，Options对象最后会封装用户设置的所有命令行信息
        //这里我要多解释一下，现在我们还没有实现Proxy模块访问Namesrv功能，但以后肯定要实现，而Proxy要访问Namesrv就肯定要知道Namesrv的IP地址和端口号
        //这个Namesrv的IP地址和端口号也是可以设置到命令行参数中的，就是使用 -n 来设置的，就像这样 -n 127.0.0.1:9876，这样Proxy模块就知道要访问的Namesrv的IP地址和端口号了
        //但是在当前方法中，并没有创建用来封装Namesrv的Option对象，这是因为在执行ServerUtil.buildCommandlineOptions(new Options())这行代码的时候，ServerUtil.buildCommandlineOptions方法中已经创建了一个Option对象
        //这个Option对象关联的命令行参数就是-n，也就是说，ServerUtil.buildCommandlineOptions方法中已经创建了一个用来封装Namesrv网络地址的Option对象并且加入到了Options对象中了
        Options options = ServerUtil.buildCommandlineOptions(new Options());
        //定义一个Option对象，用来封装用户设置得Broker配置文件路径信息
        Option opt = new Option("bc", "brokerConfigPath", true, "Broker config file path for local mode");
        opt.setRequired(false);
        options.addOption(opt);
        //定义一个Option对象，用来封装用户设置得Proxy配置文件路径信息
        opt = new Option("pc", "proxyConfigPath", true, "Proxy config file path");
        opt.setRequired(false);
        options.addOption(opt);
        //定义一个Option对象，用来封装用户设置得Proxy启动模式信息
        opt = new Option("pm", "proxyMode", true, "Proxy run in local or cluster mode");
        opt.setRequired(false);
        options.addOption(opt);
        return options;
    }

    private static void initThreadPoolMonitor() {
        ProxyConfig config = ConfigurationManager.getProxyConfig();
        //为线程池监控组件设置配置信息
        ThreadPoolMonitor.config(
                //记录线程池监控信息的日志对象
                LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME),
                //记录线程池水位线信息的日志对象
                LoggerFactory.getLogger(LoggerName.PROXY_WATER_MARK_LOGGER_NAME),
                //设置线程池是否开启打印线程栈信息
                config.isEnablePrintJstack(),
                //设置日志记录线程栈信息的时间间隔
                config.getPrintJstackInMillis(),
                //设置日志记录线程池状态信息的时间间隔
                config.getPrintThreadPoolStatusInMillis());
        //执行初始化操作
        ThreadPoolMonitor.init();
    }


}